package com.peng.sparktest.sparkstreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamTest01_Receiver01 {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("stream_receiver")
      //这里的2，表示在local模式下，指定的开启线程的个数。在stream中，最小限度也要开启2个，
      // 因为一个是用于receive job的task用于接收数据
      //一个用于处理batch时的job的task
      //当然，如果batch数据量比较大，涉及到分区出多个task并行执行，可以调高线程数，因为此时只有1个线程的话，多个task会变成串行执行
      .setMaster("local[2]")
    val context = new StreamingContext(conf, Seconds(5)) //时间间隔去触发job

    context.sparkContext.setLogLevel("ERROR")

    //透过这个的第三个参数，可以看出，此时拉取回的数据是存到了spark的blockmanager里
    //默认值是MEMORY_AND_DISK_SER_2   序列化放置到内存和磁盘
    //所有历史数据不会被删除，但会随着内存的空间不足，陆续从内存中溢写到磁盘
    val dStream: ReceiverInputDStream[String] = context.socketTextStream("localhost", 8888)

    //正常处理
    //    val result: DStream[(String, String)] = dStream.map(_.split(" ")).map(vars => {
    //      (vars(0), vars(1))
    //    })

    //模拟对一批数据处理缓慢的情况
    // 可以看出，会导致下一批得到的批数据阻塞等待前一批处理完才行
    //因此就会导致后续堆积的 批数据 越来越多
    //需要合理进行调控 间隔触发拉取job和处理job 保证拉取的数据能被消费掉，不让任务堆积
    val result: DStream[(String, String)] = dStream.map(_.split(" ")).map(vars => {
      Thread.sleep(20000)
      (vars(0), vars(1))
    })

    result.print() //输出算子 相当于去调用一次上述的job代码的action算子

    context.start() //开始执行调度

    context.awaitTermination() //阻塞等待直到程序被结束

  }

}
